#include "app_pool.h"

typedef struct
{
    void (*task)(void *);
    void *data;
} app_pool_task_t;

typedef struct
{
    pthread_t *threads;
    int pool_size;

    mqd_t mq;
    char *mq_name;

} app_pool_t;

static app_pool_t my_pool;

void *app_pool_thread(void *arg);

/// @brief 初始化线程池
/// @param pool_size
/// @return
Com_state_t app_pool_init(int pool_size)
{

    // 创建消息队列
    my_pool.pool_size = pool_size;
    my_pool.mq_name = "/pool";
    //mq_unlink(my_pool.mq_name);
    struct mq_attr attr;
    attr.mq_flags = 0;
    attr.mq_maxmsg = pool_size;
    attr.mq_msgsize = sizeof(app_pool_task_t);
    my_pool.mq = mq_open(my_pool.mq_name, O_CREAT | O_RDWR, 0666, &attr);
    if (my_pool.mq == -1)
    {
        log_error("消息队列创建失败");
        return Com_FAIL;
    }
    log_info("消息队列创建成功");

    // 分配内存，存储线程id
    my_pool.threads = malloc(sizeof(pthread_t) * pool_size);

    for (size_t i = 0; i < pool_size; i++)
    {
        if (pthread_create(&my_pool.threads[i], NULL, app_pool_thread, NULL) == -1)
        {
            log_error("线程创建失败");
            return Com_FAIL;
        }
        log_info("线程初始化成功");
    }
    return Com_OK;
}

/// @brief 接收任务
/// @param arg
/// @return
void *app_pool_thread(void *arg)
{
    while (1)
    {
        app_pool_task_t task;
        ssize_t len = mq_receive(my_pool.mq, (char *)&task, sizeof(task), NULL);
        if (len > 0)
        {
            if (task.task == NULL && task.data == NULL) // 收到退出信号
            {
                log_info("线程安全退出");
                break;
            }
            task.task(task.data);
        }
    }
    return NULL;
}

/// @brief 销毁线程池
/// @param
void app_pool_deinit(void)
{

    // 给每个线程发送一个“退出任务”，让它们主动退出
    for (size_t i = 0; i < my_pool.pool_size; i++)
    {
        app_pool_task_t exit_task = {.task = NULL, .data = NULL};
        while (mq_send(my_pool.mq, (char *)&exit_task, sizeof(exit_task), 0) == -1)
        {
            if (errno == EAGAIN)
            {
                // 队列满了，稍等一下再发
                usleep(1000);
                continue;
            }
            else
            {
                log_error("发送退出任务失败: %s", strerror(errno));
                break;
            }
        }
    }

    // 等待所有线程正常退出
    for (size_t i = 0; i < my_pool.pool_size; i++)
    {
        pthread_join(my_pool.threads[i], NULL);
    }

    // 释放资源
    free(my_pool.threads);

    my_pool.threads = NULL;

    mq_close(my_pool.mq);
    mq_unlink(my_pool.mq_name);

    log_info("线程池已经安全销毁");
}

/// @brief 添加任务 to 线程池
/// @param task 任务函数
/// @param data 传入数据
/// @return
Com_state_t app_pool_add_task(void (*task)(void *), void *data)
{
    if (task == NULL)
    {
        log_error("执行任务函数不能为空");
        return Com_FAIL;
    }

    app_pool_task_t task_info = {
        .task = task,
        .data = data};

    if (mq_send(my_pool.mq, (char *)&task_info, sizeof(app_pool_task_t), 0) == -1)
    {
        log_error("发送队列消息失败");
        return Com_FAIL;
    }
    log_info("发送队列消息成功");
    return Com_OK;
}